package com.zws.cucumber.spring.configuration;

import com.zws.cucumber.ZwsConstants;
import com.zws.cucumber.ZwsUtils;
import org.apache.commons.lang3.SystemUtils;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.EmbeddedKafkaZKBroker;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.util.Map;

/**
 * author: zws
 */
@Configuration
@Order(Ordered.HIGHEST_PRECEDENCE)
public class ZwsKafkaConf implements ApplicationContextAware {

    private ApplicationContext ctx;

    @Bean(destroyMethod = "destroy")
    @Lazy(false)
    public EmbeddedKafkaBroker embeddedKafkaBroker() {
        EmbeddedKafkaZKBroker broker = new AsyncInitEmbeddedKafkaBroker(1, true, 2);
        broker.kafkaPorts(ZwsConstants.KAFKA_PORT);
        broker.brokerProperty("log.cleaner.enable", false);

        // inject it, does it work?
        injectEnv();
        return broker;
    }

    @Bean
    @DependsOn({"embeddedKafkaBroker","syncOnResourceAsyncInitPhase"})
    public KafkaAdminClient kafkaAdminClient(EmbeddedKafkaBroker embeddedKafkaBroker) {
        return (KafkaAdminClient) AdminClient.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafkaBroker.getBrokersAsString()));
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.ctx = applicationContext;
    }

    public static class AsyncInitEmbeddedKafkaBroker extends EmbeddedKafkaZKBroker {

        public AsyncInitEmbeddedKafkaBroker(int count) {
            super(count);
        }

        public AsyncInitEmbeddedKafkaBroker(int count, boolean controlledShutdown, String... topics) {
            super(count, controlledShutdown, topics);
        }

        public AsyncInitEmbeddedKafkaBroker(int count, boolean controlledShutdown, int partitions, String... topics) {
            super(count, controlledShutdown, partitions, topics);
        }

        @Override
        public void afterPropertiesSet() {
            ZwsCommonConf.asyncInitResource(super::afterPropertiesSet);
        }
    }

    static void injectEnv() {
        if (SystemUtils.IS_OS_WINDOWS) {
            injectWindowsEnv();
        } else {
            injectUnixEnv();
        }
    }

    static void injectWindowsEnv() {
        try {
            Class<?> clazz = Class.forName("java.lang.ProcessEnvironment");

            Map<String, String> theEnvironment = ZwsUtils.getFieldValue(clazz, "theEnvironment", null);
            theEnvironment.put("env_key_1", "127.0.0.1:" + ZwsConstants.KAFKA_PORT);

            Map<String, String> theCaseInsensitiveEnvironment = ZwsUtils.getFieldValue(clazz, "theCaseInsensitiveEnvironment", null);
            theCaseInsensitiveEnvironment.put("env_key_2", "127.0.0.1:" + ZwsConstants.KAFKA_PORT);

        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
    }

    static void injectUnixEnv() {
        try {
            Class<?> clazzOfEnv = Class.forName("java.lang.ProcessEnvironment");
            Class<?> clazzOfVar = Class.forName("java.lang.ProcessEnvironment$Variable");
            Class<?> clazzOfVal = Class.forName("java.lang.ProcessEnvironment$Value");

            Method methodOfVarValueOf = ReflectionUtils.findMethod(clazzOfVar, "valueOf", String.class);
            methodOfVarValueOf.setAccessible(true);

            Method methodOfValValueOf = ReflectionUtils.findMethod(clazzOfVal, "valueOf", String.class);
            methodOfValValueOf.setAccessible(true);

            Map theEnvironment = ZwsUtils.getFieldValue(clazzOfEnv, "theEnvironment", null);
            theEnvironment.put(
                    methodOfVarValueOf.invoke(clazzOfVar, "env_key_1"),
                    methodOfValValueOf.invoke(clazzOfVal, "127.0.0.1:" + ZwsConstants.KAFKA_PORT));

        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
    }


}
